package keter.rabbitmq.topic;

import com.rabbitmq.client.*;

import java.io.IOException;

public class TopicPersistRecv {

  private final static String EXCHANGE_NAME = "topic_exchange";
  private final static String QUEUE_NAME = "topic_queue";
  private final static String ROUTING_KEY = "topic.bind";

  public static void main(String[] argv) throws Exception {
    ConnectionFactory factory = new ConnectionFactory();
    factory.setHost("192.168.1.100");
    factory.setUsername("haha");
    factory.setPassword("haha");
    Connection connection = factory.newConnection();
    Channel channel = connection.createChannel();

    channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.TOPIC);

    channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, ROUTING_KEY);

    System.out.println(" [*] Waiting for messages. To exit press CTRL+C");

    //保证在接收端一个消息没有处理完时不会接收另一个消息，即接收端发送了ack后才会接收下一个消息。
    //在这种情况下发送端会尝试把消息发送给下一个not busy的接收端。
    channel.basicQos(1);

    Consumer consumer = new DefaultConsumer(channel) {
      @Override
      public void handleDelivery(String consumerTag, Envelope envelope,
                                 AMQP.BasicProperties properties, byte[] body) throws IOException {
        String message = new String(body, "UTF-8");
        System.out.println(" [x] Received '" + envelope.getRoutingKey() + "':'" + message + "'");
        //如果channel.basicConsume的autoAck为false，则需处理消息后主动通知给RabbitMQ
        channel.basicAck(envelope.getDeliveryTag(), false);
        System.out.println(" [x] ACK send! ");
      }
    };
    channel.basicConsume(QUEUE_NAME, false, consumer);
  }
}

